package com.study.bigdata.spark.core.rdd.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Scala16_RDD_Operator_Transform{
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    conf.set("spark.local.dir","D:\\hadoopbook\\spark\\test")
    val sc = new SparkContext(conf)
    // TODO 算子 - 转换 -  aggregateByKey
    // 取出每个分区内相同key的最大值然后分区间相加
    // [(a,1),(a,2),(b,3)]
    //    =>[(a,2),(b,3)]
    //          =>[(a,8),(b,8)]
    //    =>[(b,5),(a,6)]
    // [(b,4),(b,5),(a,6)]
    val rdd = sc.makeRDD(
      List(
        ("a", 1), ("a", 2), ("b", 3),
        ("b", 4), ("b", 5), ("a", 6)
      ), 2
    )
    // aggregateByKey算子是函数柯里化，存在两个参数列表
    /*
    1. 第一个参数列表中的参数表示初始值
    2. 第二个参数列表中含有两个参数
      2.1 第一个参数表示分区内的计算规则
      2.2 第二个参数表示分区间的计算规则
     */
    val rdd1 = rdd.aggregateByKey(0)(
      (x, y) => {
        math.max(x, y)
      },
      (x, y) => {
        x + y
      }
    )
    rdd1.collect().foreach(println)
    /*
    (b,8)
    (a,8)
     */

    sc.stop()

  }

}
